package com.example.lkfy.Interceptor;

import com.example.lkfy.model.MyConsumer;
import com.example.lkfy.server.RedisConsumer;
import com.example.lkfy.utils.RedisMqUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
public class RedisMqRunner implements ApplicationRunner {
    @Autowired
    private RedisMqUtil redisMqUtil;
    /**
     * 主题和订阅关系
     */
    private static final ConcurrentHashMap<String, RedisConsumer> topicMap = new ConcurrentHashMap<>();

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    private ApplicationContext context;

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        Map<String, Object> map = context.getBeansWithAnnotation(MyConsumer.class);
        map.values().forEach(value->{
            if(!(value instanceof RedisConsumer)){
                log.warn("检测到被@MyConsumer注解的类未实现RedisConsumer", value.getClass().getCanonicalName());
                return;
            }
            MyConsumer[] myConsumers = value.getClass().getAnnotationsByType(MyConsumer.class);
            if(Objects.isNull(myConsumers))
            {
                log.warn("未检测到被@MyConsumer注解的类", value.getClass().getCanonicalName());
                return;
            }
            MyConsumer myConsumer = myConsumers[0];
            String topic = myConsumer.topic();
            if(topicMap.containsKey(topic)){
                log.error("多个消费者{},消费同一个消息:{},已忽略", value.getClass().getCanonicalName(), topic);
            } else {
                topicMap.put(topic, (RedisConsumer) value);
            }
        });
        log.info("redis消息订阅成功");
        threadPoolExecutor.execute(loop());
    }

    /**
     * 当个线程死循环执行消息监听
     * @return
     */
    private Runnable loop(){
        return ()->{
          while (true){
              AtomicInteger count = new AtomicInteger(0);
              topicMap.forEach((k,consumer)->{
                  try {
                      Object message =  redisMqUtil.rightPop(k);
                      if(Objects.isNull(message)){
                          count.getAndIncrement();
                      }else{
                          dealTask(consumer, k, message);
                      }
                  }catch (RedisConnectionFailureException connException) {
                      log.error("redis无法连接,5s后重试", connException);
                      sleep(5);
                  } catch (Exception e) {
                      log.error("redis消息队列异常", e);
                  }
              });
              if(count.get() == topicMap.keySet().size()){
                  sleep(1);//休眠一秒
              }
          }
        };
    }

    /**
     * 消费者消费消息
     * @param consumer
     * @param k
     * @param v
     */
    private void dealTask(RedisConsumer consumer, Object k, Object v){
        threadPoolExecutor.execute(()->{
            try {
                consumer.deal(v);
            }catch (Exception e){
                log.error("执行消费任务出错", e);
                //非广播消息进行数据回补
                redisMqUtil.rightPush(k, v);
            }
        });
    }

    /**
     * 休眠时间
     * @param s
     */
    private void sleep(int s) {
        try {
            TimeUnit.SECONDS.sleep(s);
        } catch (Exception e) {
            log.error("休眠失败", e);
        }
    }
}
